package com.chat.transmission.service.impl;

import cn.hutool.json.JSONUtil;
import com.chat.common.common.constant.RedisKeyConstant;
import com.chat.common.dto.MessageDTO;
import com.chat.common.enums.NettyUdpPackageType;
import com.chat.common.vo.GroupVO;
import com.chat.common.vo.MessageVO;
import com.chat.common.vo.NettyUdpVo;
import com.chat.transmission.common.UdpPacket;
import com.chat.transmission.event.event.UserGetMsgEvent;
import com.chat.transmission.event.event.UserSendMsgEvent;
import com.chat.transmission.event.source.GetMsgEventSource;
import com.chat.transmission.event.source.SendMsgEventSource;
import com.chat.transmission.service.BusinessService;
import com.chat.transmission.service.RedisService;
import com.chat.transmission.util.RedisHashUtil;
import com.google.gson.Gson;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.CharsetUtil;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author wei kunJie
 * @date 2024/3/22 20:34:03
 */
@Service
public class RedisServiceImpl implements RedisService {
    private final RedisTemplate<String, Object> redisTemplate;
    private final RedisHashUtil redisHashUtil;
    private final ConcurrentHashMap<String, Object> getMsgEventSourceMap;
    private final Gson gson;
    private final BusinessService businessService;


    public RedisServiceImpl(RedisTemplate<String, Object> redisTemplate, RedisHashUtil redisHashUtil, ConcurrentHashMap<String, Object> getMsgEventSourceMap, Gson gson, BusinessService businessService) {
        this.redisTemplate = redisTemplate;
        this.redisHashUtil = redisHashUtil;
        this.getMsgEventSourceMap = getMsgEventSourceMap;
        this.gson = gson;
        this.businessService = businessService;
    }

    /**
     * 监听用户发送消息的事件
     *
     * @param userSendMsgEvent
     */
    @Override
    @EventListener
    @Async("taskExecutor")//指定线程池
    public void onUserSendMsgEvent(UserSendMsgEvent userSendMsgEvent) {
        SendMsgEventSource sendMsgEventSource = (SendMsgEventSource) userSendMsgEvent.getSource();
        MessageDTO messageDTO = sendMsgEventSource.getMessageDTO();
        Long ackMsgId = messageDTO.getId();
        //发送，存到redis中
        redisHashUtil.publishMessage(messageDTO);
        //响应客户端，对用户来说，服务器接收到，表示已发送
        //如果客户端发送后没有在超时前收到Ack就是发送失败（红色感叹号）
        //给用户回复已完成发送 用Netty，加ico去
        ChannelHandlerContext ctx = sendMsgEventSource.getCtx();
        InetSocketAddress sender = sendMsgEventSource.getUserInetSocketAddress();
        //TODO：这里response传UdpVO json
        NettyUdpVo udpVo = new NettyUdpVo();
        udpVo.setType(NettyUdpPackageType.SEND_MSG_RESPONSE);
        udpVo.setData(NettyUdpVo.NettyUdpDataBuilder.builder()
                .put("ackMessageId", ackMsgId)//ack
                .build());
        String response = gson.toJson(udpVo);//把返回体转成json
        DatagramPacket data = new DatagramPacket(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8), sender);
        //发送Udp包
        ctx.writeAndFlush(data);
    }

    /**
     * 监听用户心跳事件
     *
     * @param userGetMsgEvent
     */
    @Override
    @EventListener
    @Async("taskExecutor")
    public void onUserGetMsgEvent(UserGetMsgEvent userGetMsgEvent) {
        GetMsgEventSource userGetMsgEventSource = (GetMsgEventSource) userGetMsgEvent.getSource();
        //给用户发他要获取的未读消息
        ChannelHandlerContext ctx = userGetMsgEventSource.getCtx();
        InetSocketAddress sender = userGetMsgEventSource.getUserInetSocketAddress();
        //从Redis中获取未读消息
        String account = (String) redisTemplate.opsForValue().get(RedisKeyConstant.TOKEN_ACCOUNT + userGetMsgEventSource.getToken());
        final List<MessageVO> messageVOList = redisHashUtil.consumeMessage(account);
        //从Redis中获取未读通知
        Long userId = (Long) redisTemplate.opsForValue().get(RedisKeyConstant.TOKEN_USER + userGetMsgEventSource.getToken());
        Map<String, String> noticesJson = redisHashUtil.consumeNotice(userId);
        //获取用户所在的所有群
        List<GroupVO> userGroups = businessService.getUserGroups(userId);
        //获取群信息
        if (userGroups != null) {
            //记录当前时间
            long millis = System.currentTimeMillis();
            //将消息过滤，只要用户没有获取过的
            userGroups.forEach(groupVO -> {
                List<MessageVO> userGroupMessages = redisHashUtil.getGroupMessage(groupVO.getGroupAccount());
                if (userGroupMessages != null) {
                    messageVOList.addAll(userGroupMessages.stream()
                            .filter(it -> !it.getFromUserAccount().equals(account))//不是这个用户发的
                            .filter(it -> it.getTimeMillis() > groupVO.getGroupMsgGetLastTimeMilli())//获取时间在上一次获取时间之后
                            .collect(Collectors.toList())
                    );
                }
            });
            //更新“上一次获取消息的时间”（对下一次来说本次就是上一次）
            businessService.updateUserGroupsCache(userGroups.stream()
                            .map(it -> it.updateGroupMsgGetLastTimeMilli(millis))
                            .collect(Collectors.toList())
                    , userId);

        }
        if ((messageVOList.size() == 0) && (noticesJson == null || noticesJson.size() == 0)) {
            // 没有这个用户的未读消息或通知
            // 创建DatagramPacket
            String response = "noMessage:" + account;
            DatagramPacket packet = new DatagramPacket(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8), sender);
            // 发送UDP包
            ctx.writeAndFlush(packet);
            //移除用户正在获取数据状态
            if (getMsgEventSourceMap.contains(userGetMsgEventSource.getToken()))
                getMsgEventSourceMap.remove(userGetMsgEventSource.getToken());
            return;
        }
        Map<String, Object> noticesObj = new HashMap<>();
        //判空后，把json字符串转成对象，解决直接传json字符串传给客户端json字符串被转义的问题
        if (noticesJson != null) {
            noticesJson.forEach((s, s2) -> noticesObj.put(s, JSONUtil.parseObj(s2)));
        }
        //把dto转成vo
        NettyUdpVo udpVo = new NettyUdpVo();
        udpVo.setData(NettyUdpVo.NettyUdpDataBuilder.builder()
                .put("messageVOList", messageVOList)//向VO中放未读消息
                .put("notices", noticesObj)//向VO中放未读通知
                .build());
//        udpVo.setId();
        udpVo.setType(NettyUdpPackageType.GET_MSG_RESPONSE);
        // 将要发送的数据
        String dataToSend = gson.toJson(udpVo);
        byte[] dataBytes = dataToSend.getBytes(CharsetUtil.UTF_8);
        int segmentSize = 1024; // 每个分段的字节数
        int totalSegments = (int) Math.ceil((double) dataBytes.length / segmentSize); // 计算总共的分段数
        // 循环发送每个数据片段
        for (int i = 0; i < totalSegments; i++) {
            int startIndex = i * segmentSize;
            int endIndex = Math.min(startIndex + segmentSize, dataBytes.length);
            String segment = new String(dataBytes, startIndex, endIndex - startIndex);
            UdpPacket udpPacket = new UdpPacket();
            udpPacket.setId((long) i);// 设置标志位，用于识别片段顺序等信息
            udpPacket.setData(segment);
            udpPacket.setTotal(totalSegments);//一共有多少个包
            // 创建DatagramPacket
            String response = gson.toJson(udpPacket);
            DatagramPacket packet = new DatagramPacket(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8), sender);
            // 发送UDP包
            ctx.writeAndFlush(packet);
        }
        //移除用户正在获取数据状态
        if (getMsgEventSourceMap.contains(userGetMsgEventSource.getToken()))
            getMsgEventSourceMap.remove(userGetMsgEventSource.getToken());
    }

}
